package com.taotao.job.executor.core;

import com.google.common.base.Preconditions;
import com.taotao.job.core.runresult.RunRate;
import com.taotao.job.rpc.core.utils.StringUtils;
import lombok.extern.slf4j.Slf4j;

import java.util.HashMap;
import java.util.Objects;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * 运行进度缓存
 *
 * @author fanbeibei
 * @date 2021/4/7 21:25
 */
@Slf4j
public class RunRateCache {
    private static RunRateCache runRateCache;

    private RunRateCache() {

    }

    public static RunRateCache getInstance() {
        if (null == runRateCache) {
            synchronized (RunRateCache.class) {
                if (null == runRateCache) {
                    runRateCache = new RunRateCache();
                    runRateCache.init();
                }
            }
        }

        return runRateCache;
    }


    /**
     * jobInstanceId -> runRate
     */
    protected final ConcurrentHashMap<Long, String> rateCache = new ConcurrentHashMap<>();

    protected final BlockingQueue<RunRate> queue = new ArrayBlockingQueue<>(10000);


    protected volatile boolean stoped = false;

    private void init() {
        new Thread(new Runnable() {
            @Override
            public void run() {
                while (stoped) {
                    try {
                        //串行化处理保证结果正确
                        RunRate runRate = queue.poll(5, TimeUnit.SECONDS);
                        if (null != runRate) {
                            changeRate(runRate.getRunInstanceId(), runRate.getRate());
                        }
                    } catch (InterruptedException e) {
                        break;
                    }
                }
            }
        }).start();
    }

    private void changeRate(Long runInstanceId, String rate) {
        String rate0 = rateCache.get(runInstanceId);

        if (RunRate.isFinalRate(rate0)) {
            log.info(" ignore  runInstanceId={},rate={}", runInstanceId, rate);
            return;
        }

        rateCache.put(runInstanceId, rate);
    }

    public void putRunRate(Long runInstanceId, String runRate) {
        Preconditions.checkArgument(null != runInstanceId);
        Preconditions.checkArgument(StringUtils.isNotBlank(runRate));

        queue.add(new RunRate(runInstanceId, runRate));
    }

    public String getRunRate(Long runInstanceId) {
        return rateCache.get(runInstanceId);
    }

    public HashMap<Long, String> getAllRunRate() {
        return new HashMap<>(rateCache);
    }
}
